package com.yuanfeng.goods.util.es;

import com.alibaba.fastjson.JSON;
import com.yuanfeng.commoms.constant.CommonType;
import com.yuanfeng.commoms.dto.shop.ArticleDTO;
import com.yuanfeng.commoms.rocketmq.config.JmsConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author Stranger
 * @date 2023/5/8 16:26
 */
@Slf4j
@Component
@RefreshScope
public class ESMqConsumer {
    private DefaultMQPushConsumer consumer;

    @Value("${rocketmq.name-server}")
    private String nameServer;

    @PostConstruct
    public void ini() throws MQClientException {

        consumer = new DefaultMQPushConsumer(JmsConfig.DEFAULT_ES_TOPIC);
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 默认是集群方式，可以更改为广播，但是广播方式不支持重试
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.subscribe(JmsConfig.DEFAULT_ES_TOPIC, "*");


        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            MessageExt msg = msgs.get(0);

            int times = msg.getReconsumeTimes();
            if (times > 0) {
                log.info("重试次数=" + times);
            }
            String topic = msg.getTopic();
            String body = new String(msg.getBody(), StandardCharsets.UTF_8);
            String tags = msg.getTags();
            String keys = msg.getKeys();
            log.info("消费消息:" + "topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);
            Map<String, Object> map = JSON.parseObject(body);
            Integer type = (Integer) map.get("type");
            String str = map.get("commonIdS").toString().replace("[", "").replace("]", "");
            String[] list = str.split(",");
            List<ArticleDTO> articles = new ArrayList<>();
            List<Integer> list2 = new ArrayList<>();
            for (String s : list) {
                ArticleDTO article = new ArticleDTO();
                article.setCommonId(Integer.valueOf(s));
                articles.add(article);
                list2.add(Integer.parseInt(s));
            }
            if (CommonType.operateStatus.ADD.getCode().equals(type)) {
                //更新es的商品数据
                ESUtils.addOrUpdate(articles);
            } else {
                ESUtils.delete(list2);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();
        log.info("consumer start ...");
    }

}
